package com.shujia.flink.tf


import org.apache.flink.streaming.api.scala._

object Demo2Agg {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val student: DataStream[String] = env.readTextFile("data/students.txt")

    val students: DataStream[Student] = student.map(line => {
      val split: Array[String] = line.split(",")
      val id: String = split(0)
      val name: String = split(1)
      val age: Int = split(2).toInt
      val gender: String = split(3)
      val clazz: String = split(4)
      Student(id, name, age, gender, clazz)
    })

    students.keyBy(_.clazz)
      .sum(2)//指定下标聚合
//      .print()

    /**
      * KeyedStream.sum(0)
      * KeyedStream.sum("key")
      * KeyedStream.min(0)
      * KeyedStream.min("key")
      * KeyedStream.max(0)
      * KeyedStream.max("key")
      * KeyedStream.minBy(0)
      * KeyedStream.minBy("key")
      * KeyedStream.maxBy(0)
      * KeyedStream.maxBy("key")
      *
      *
      * max 和 maxBy 之间的区别在于 max 返回流中的最大值，但 maxBy 返回具有最大值的键， min 和 minBy 同理。
      *
      */

    students
        .keyBy(_.clazz)
        .min("age")
        .print()



    env.execute()
  }
}
case class Student(id: String, name: String, age: Int, gender: String, clazz: String)